package com.seventeen.monster.redisson.support;

import com.seventeen.monster.common.utils.StringUtils;
import com.seventeen.monster.redisson.annotation.DelayedQueueListener;
import com.seventeen.monster.redisson.core.DelayedQueueHandler;
import com.seventeen.monster.redisson.core.DelayedQueueUtils;
import com.seventeen.monster.redisson.core.Message;
import com.seventeen.monster.redisson.core.SleepUtils;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.SmartLifecycle;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author seventeen
 */
public class DefaultDelayedQueueListenerContainer implements DelayedQueueListenerContainer,
        SmartLifecycle {
    private final static Logger log = LoggerFactory.getLogger(DefaultDelayedQueueListenerContainer.class);
    private volatile boolean running;
    private String name;
    private String queueName;
    private int consumeThreadNumber;
    private int retryLevelWhenNextConsume;
    private DelayedQueueHandler delayedQueueHandler;
    private RedissonClient redissonClient;
    private RBlockingQueue blockingQueue;

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public void start() {
        if (this.isRunning()) {
            throw new IllegalStateException("Redisson delay queue container already running. " + this);
        }
        this.blockingQueue = redissonClient.getBlockingQueue(queueName);
        this.setRunning(true);
        for (int i = 1; i <= this.consumeThreadNumber; i++) {
            DelayedQueueTask delayedQueueTask = new DelayedQueueTask(queueName + "-" + i);
            Thread thread = new Thread(delayedQueueTask);
            thread.setDaemon(true);
            thread.start();
        }
        log.info("Running redisson delay queue container: {}", this);
    }

    public DelayedQueueHandler getDelayedQueueHandler() {
        return delayedQueueHandler;
    }

    public void setDelayedQueueHandler(DelayedQueueHandler delayedQueueHandler) {
        this.delayedQueueHandler = delayedQueueHandler;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public RedissonClient getRedissonClient() {
        return redissonClient;
    }

    public void setRedissonClient(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    @Override
    public void stop() {
        this.setRunning(false);
    }

    @Override
    public boolean isRunning() {
        return running;
    }

    private void setRunning(boolean running) {
        this.running = running;
    }

    public void setDelayedQueueListener(DelayedQueueListener anno) {
        this.queueName = anno.queueName();
        this.consumeThreadNumber = anno.consumeThreadNumber();
        this.retryLevelWhenNextConsume = anno.retryLevelWhenNextConsume();
    }

    @Override
    public void destroy() throws Exception {
        if (this.isRunning()) {
            log.info("Redisson delay queue container stop, {}", this);
            this.setRunning(false);
            SleepUtils.sleepSeconds(3);
        }
    }

    private class DelayedQueueTask implements Runnable {
        public DelayedQueueTask(String name) {
            this.name = name;
        }

        private String name;

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            log.info("Redisson delay queue {} run", queueName);
            while (isRunning()) {
                Message message;
                try {
                    message = (Message) blockingQueue.poll(1, TimeUnit.SECONDS);
                } catch (Exception e) {
                    log.error("Redisson delay queue take exception, ", e);
                    SleepUtils.sleep(200);
                    continue;
                }
                if (Objects.nonNull(message)) {
                    try {
                        delayedQueueHandler.onMessage(message.getPayload());
                    } catch (Exception e) {
                        log.error("Redisson delay queue {} message {} handler exception, ", queueName, message, e);
                        // 判断是否重试
                        try {
                            if (retryLevelWhenNextConsume > 0) {
                                Map<String, String> headers = message.getHeaders();
                                if (headers == null) {
                                    headers = new HashMap<>(16);
                                }
                                // 重试次数
                                String retryCount = headers.get("retryCount");
                                int retryLevel;
                                if (StringUtils.isBlank(retryCount)) {
                                    retryLevel = retryLevelWhenNextConsume;
                                } else {
                                    int count = Integer.parseInt(retryCount);
                                    retryLevel = count + 1;
                                }
                                if (retryLevel > 18) {
                                    log.error("Redisson delay queue retry time is too many, retry times is 18, queue {}, message {}", queueName, message);
                                } else {
                                    headers.put("retryCount", String.valueOf(retryLevel));
                                    message.setHeaders(headers);
                                    DelayedQueueUtils.retry(queueName, message, retryLevel);
                                }
                            }
                        } catch (Exception ex) {
                            log.error("Redisson delay queue {} message {} retry exception, ", queueName, message, e);
                        }
                    }
                    log.info("Redisson delay queue {} message {} handler success", queueName, message);
                }
            }
            log.info("Redisson delay queue task {} stop", this.getName());
        }
    }
}
